package com.lujieni.kafka.message;

import com.alibaba.fastjson.JSON;
import com.lujieni.kafka.dto.OrderDTO;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class KafkaConsumer {
    /**
     * 设置手动提交偏移量
     *
     * @param list
     */
  /*  @KafkaListener(topics = "lujieni")
    public void setCommitType(List<ConsumerRecord<String, String>> list, Acknowledgment ack) {
        for (ConsumerRecord<String, String> record: list) {
            System.out.println("setCommitType");
            System.out.println(Thread.currentThread().getName()+" 内容：" + record.value());
            System.out.println("分区：" + record.partition());
            System.out.println("偏移量：" + record.offset());
            System.out.println("创建消息的时间戳：" + record.timestamp());
        }
        System.out.println("aa");
        ack.acknowledge();

    }*/


    /**
     * 设置手动提交偏移量
     * 实体类是json格式的字符串，消费者再转下即可
     * @param
     */
    @KafkaListener(topics = "idea")
    public void lujeiniConsumerOneByOne(ConsumerRecord<String,String> record, Acknowledgment ack) {
        try {
            System.out.println("lujeiniConsumerOneByOne");
            System.out.println(Thread.currentThread().getName()+" 内容：" + JSON.parseObject(record.value(), OrderDTO.class).toString());
            //System.out.println(Thread.currentThread().getName()+" 内容：" + record.value());
            System.out.println("分区：" + record.partition());
            System.out.println("偏移量：" + record.offset());
            System.out.println("创建消息的时间戳：" + record.timestamp());
        } catch (Exception e) {

        } finally {
            ack.acknowledge();
        }
    }

}
